package vn.ebuzz.sms.worker;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import vn.ebuzz.sms.core.Configuration;
import vn.ebuzz.sms.core.Database;
import vn.ebuzz.sms.core.Incoming;
import vn.ebuzz.sms.core.Outgoing;
import vn.ebuzz.sms.core.Util;

public class ProcessQueue {
	private static Logger log = Logger.getLogger("procqueue");	

	private static ProcessQueue q = null;
	private HashMap<String, String> cache;

	volatile boolean isRunning = true;

	private ProcessQueue(){
		cache = new HashMap<String, String>();
		scheduler = Executors.newSingleThreadScheduledExecutor();
		for(int i = 0; i < Configuration.MAX_THREAD; i ++){
			ProcessWorker t = new ProcessWorker(i);
			t.start();
			log.info("Start thread " + t.getId());
			// sleep between thread
			try {
				Thread.sleep(Configuration.SEPERATOR);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}			
		}
	}

	public static ProcessQueue getInstance(){
		if(q == null)
			q = new ProcessQueue();
		return q;
	}

	public void stop(){
		isRunning = false;
		cache.clear();
		scheduler.shutdownNow();
	}

	private ScheduledExecutorService scheduler;

	private void insertCache(final String keyword, String message){
		cache.put(keyword, message);
		scheduler.schedule(new Runnable() {			
			public void run() {
				cache.remove(keyword);
			}
		}, Configuration.CACHE, TimeUnit.MILLISECONDS);
	}

	private class ProcessWorker extends Thread{
		private int index;
		public ProcessWorker(int i){
			this.index = i;
		}
		@Override
		public void run() {
			long start = System.currentTimeMillis();
			long threadId = Thread.currentThread().getId();			
			while(isRunning){
				log.debug("ProcessWorker" + threadId + " check DB");
				ArrayList<Incoming> myInList = new ArrayList<Incoming>();

				Connection conn = Database.getConnection();
				Statement st = null;
				ResultSet rs = null;
				try{
					st = conn.createStatement();
					rs = st.executeQuery("select id, sender, receiver," +
							" message, mo_id, session_id," +
							" unique_id, status from incoming" +
							" where status = " + Configuration.MO_QUEUED +
							" and mod(id, " + Configuration.MAX_THREAD + ") = " + index +
					" order by created asc LIMIT " + Configuration.MAX_BATCH);
					while(rs.next()){
						Incoming in = new Incoming();
						in.setId(rs.getInt("id"));
						in.setSender(rs.getString("sender"));
						in.setStatus(rs.getInt("status"));
						in.setReceiver(rs.getString("receiver"));
						in.setMessage(rs.getString("message"));
						in.setMo_id(rs.getString("mo_id"));
						in.setSession_id(rs.getString("session_id"));
						in.setUnique_id(rs.getString("unique_id"));
						myInList.add(in);
					}
				} catch(Exception e){
					log.error("Error", e);
				} finally{
					if(rs != null)
						try {
							rs.close();
						} catch (SQLException e) {
							log.error("Error", e);
						}
						if(st != null)
							try {
								st.close();
							} catch (SQLException e) {
								log.error("Error", e);
							}
							if(conn != null)
								Database.closeConnection(conn);					
				}
				log.debug("ProcessWorker" + threadId + " get " + myInList.size() + " jobs");
				if(myInList.size() > 0){
					ArrayList<Outgoing> myOutList = new ArrayList<Outgoing>();

					String defaultDT = "Soan DT SBD gui 6780, trong so SBD gom MATRUONG MAKHOI SBD viet lien, de tra cuu ket qua. Cam on ban da su dung dich vu.";
					String wrongDT = "Soan DT SBD gui 6780, trong so SBD gom MATRUONG MAKHOI SBD viet lien, de tra cuu ket qua. Cam on ban da su dung dich vu.";
					String pendingDT = "Soan DT SBD gui 6780, trong so SBD gom MATRUONG MAKHOI SBD viet lien, de tra cuu ket qua. Cam on ban da su dung dich vu.";

					String defaultDH = "Soan DH MAKHOI MON MADE gui 6780 de tra cuu dap an ky thi DH 2013. Cam on ban da su dung dich vu.";
					String wrongDH = "Soan DH MAKHOI MON MADE gui 6780 de tra cuu dap an ky thi DH 2013. Cam on ban da su dung dich vu.";
					String pendingDH = "Soan DH MAKHOI MON MADE gui 6780 de tra cuu dap an ky thi DH 2013. Cam on ban da su dung dich vu.";
					
					try {
						Properties prop = Configuration.getInstance().loadFile("content.properties");
						defaultDT = prop.getProperty("DEFAULT_DT");
						wrongDT = prop.getProperty("WRONG_DT");
						pendingDT = prop.getProperty("PENDING_DT");
						
						defaultDH = prop.getProperty("DEFAULT_DH");
						wrongDH = prop.getProperty("WRONG_DH");
						pendingDH = prop.getProperty("PENDING_DH");
					} catch (FileNotFoundException e) {
						log.error("Error", e);
					} catch (IOException e) {
						log.error("Error", e);
					}

					for(int i = 0; i < myInList.size(); i ++){
						Incoming in = myInList.get(i);
						String mo = in.getMessage();

						MtWrapper tmp;
						if(mo.toUpperCase().startsWith("DH"))
							tmp = getAnswer(mo, defaultDH, pendingDH, wrongDH);
						else
							tmp = getScore(mo, defaultDT, pendingDT, wrongDT);

						String mt = tmp.mt;
						int moStatus = tmp.moStatus;

						Outgoing out = new Outgoing();
						out.setMo_id(in.getMo_id());
						out.setSession_id(in.getSession_id());
						out.setUnique_id(in.getUnique_id());
						out.setMessage(mt);
						out.setSender(in.getReceiver());
						out.setReceiver(in.getSender());
						out.setIncoming_id(in.getId());
						out.setStatus(Configuration.MT_QUEUED);
						out.setMtcount(Util.countLength(mt));
						out.setMostatus(moStatus);
						myOutList.add(out);
					}
					//TODO: batch insert
					insertRecordsIntoOutgoing(myOutList);

					long end = System.currentTimeMillis();
					long exectime = end - start;
					if(exectime < Configuration.THRESHOLD){
						try {
							Thread.sleep(exectime);
						} catch (InterruptedException e) {
							log.error("Error", e);
						}						
					}
				}
				else{
					try {
						Thread.sleep(Configuration.SLEEP);
					} catch (InterruptedException e) {
						log.error("Error", e);
					}
				}
				log.debug("ProcessWorker" + threadId + " finish " + myInList.size() + " jobs");				
			}
		}
	}

	class MtWrapper{
		public String mt;
		public int moStatus;

		public MtWrapper(String mt, int moStatus){
			this.mt = mt;
			this.moStatus = moStatus;
		}		
	}

	private MtWrapper getScore(String mo, String defaultMsg, String pendingMsg, String wrongMsg){
		String mt = defaultMsg;
		int moStatus = Configuration.MO_WRONG;

		String[] keyword = mo.split(" ");
		int length = keyword.length;

		String keyword1 = "";
		if(length > 1){
			keyword1 = keyword[1];

			String sql = "select message" +
					" from score" +
					" where upper(keyword1) = '" + keyword1 + "'";

			Connection conn = Database.getConnection();
			Statement st = null;
			ResultSet rs = null;
			boolean isFound = false;
			try{
				st = conn.createStatement();
				rs = st.executeQuery(sql);

				while(rs.next()){
					mt = rs.getString("message");
					isFound = true;
					break;
				}
			} catch(Exception e){
				log.error("Error", e);
			} finally{
				if(rs != null)
					try {
						rs.close();
					} catch (SQLException e) {
						log.error("Error", e);
					}
					if(st != null)
						try {
							st.close();
						} catch (SQLException e) {
							log.error("Error", e);
						}
						if(conn != null)
							Database.closeConnection(conn);
			}
			if(isFound){
				log.debug("Found from DB");
				moStatus = Configuration.MO_FINISHED;
			}
			else{
				log.debug("Not found from DB");
				mt = pendingMsg;
				moStatus = Configuration.MO_PENDING;
			}
		}
		else{
			mt = defaultMsg;
			moStatus = 1;
		}

		return new MtWrapper(mt, moStatus);
	}

	private MtWrapper getAnswer(String mo, String defaultMsg, String pendingMsg, String wrongMsg){
		String mt = defaultMsg;
		int moStatus = Configuration.MO_WRONG;

		//TODO: check keyword0 for service type
		// split into at max 5 keywords
		String[] keyword = mo.split(" ", 5);
		int length = keyword.length;
		String keyword1 = "";
		String keyword2 = "";
		String keyword3 = "";

		if(length > 1){
			keyword1 = keyword[1];
		}
		if(length > 2){
			keyword2 = keyword[2];
		}
		if(length > 3){
			keyword3 = keyword[3];
		}

		if(length == 1){
			// TODO: default MT
			mt = defaultMsg;
			moStatus = Configuration.MO_FINISHED;
		}
		else{
			//TODO: look in cache first
			String key = keyword1 + " " + keyword2 + " " + keyword3;
			if(cache.get(key) != null){
				log.debug("Found in cache for " + key);
				mt = cache.get(key);
			}
			else{
				log.debug("Not found in cache for " + key);
				//TODO: search DB for keyword1, keyword2, keyword3
				String sql = "select message from answer where 1 = 1";
				sql += " and upper(keyword1) = '" + keyword1 + "'";
				sql += " and upper(keyword2) = '" + keyword2 + "'";
				sql += " and upper(keyword3) = '" + keyword3 + "'";

				Connection conn2 = Database.getConnection();
				Statement st2 = null;
				ResultSet rs2 = null;
				boolean isFound = false;

				try{
					st2 = conn2.createStatement();
					rs2 = st2.executeQuery(sql);
					while(rs2.next()){
						mt = rs2.getString("message");
						isFound = true;
						break;
					}
				} catch(Exception e){
					log.error("Error", e);
				} finally{
					if(rs2 != null)
						try {
							rs2.close();
						} catch (SQLException e) {
							log.error("Error", e);
						}
						if(st2 != null)
							try {
								st2.close();
							} catch (SQLException e) {
								log.error("Error", e);
							}
							if(conn2 != null)
								Database.closeConnection(conn2);
				}
				if(!isFound){
					//TODO: not found from DB
					log.debug("Not found from DB");
					mt = wrongMsg;
					moStatus = Configuration.MO_WRONG;
				}
				else{
					if(mt.length() > 0){									
						//TODO: good mt, save to cache
						log.debug("Found from DB and insert to cache");
						insertCache(key, mt);
						moStatus = Configuration.MO_FINISHED; // finish for this mo
					}
					else{
						//TODO: blank mt, not save to cache
						log.debug("Found from DB but pending");
						mt = pendingMsg;
						moStatus = Configuration.MO_PENDING; // pending for this mo										
					}
				}
			}
		}		
		return new MtWrapper(mt, moStatus);
	}

	private void insertRecordsIntoOutgoing(List<Outgoing> listOutgoings){
		Connection conn = null;
		PreparedStatement preparedStatement = null;

		log.debug("Insert batch outgoing " + listOutgoings.size());
		String insertTableSQL = "insert into outgoing" +
		" (sender, receiver, message, status," +
		" incoming_id, mo_id, session_id, unique_id," +
		" result, mtcount, mostatus) values(?,?,?,?,?,?,?,?,?,?,?)";

		try {
			conn = Database.getConnection();
			preparedStatement = conn.prepareStatement(insertTableSQL);

			conn.setAutoCommit(false);
			for(Outgoing outgoing:listOutgoings){
				preparedStatement.setString(1, outgoing.getSender());
				preparedStatement.setString(2, outgoing.getReceiver());
				preparedStatement.setString(3, outgoing.getMessage());
				preparedStatement.setInt(4, outgoing.getStatus());
				preparedStatement.setInt(5, outgoing.getIncoming_id());
				preparedStatement.setString(6, outgoing.getMo_id());
				preparedStatement.setString(7, outgoing.getSession_id());
				preparedStatement.setString(8, outgoing.getUnique_id());
				preparedStatement.setString(9, outgoing.getResult());
				preparedStatement.setInt(10, outgoing.getMtcount());
				preparedStatement.setInt(11, outgoing.getMostatus());
				preparedStatement.addBatch();
			}
			preparedStatement.executeBatch();
			conn.commit();
			conn.setAutoCommit(true);
		} catch (SQLException e) {
			log.error("Error", e);
			try {
				conn.rollback();
			} catch (SQLException e1) {
				log.error("Error", e);
			}
		} finally {
			if (preparedStatement != null) {
				try {
					preparedStatement.close();
				} catch (SQLException e) {
					log.error("Error", e);
				}
			}
			if (conn != null) {
				Database.closeConnection(conn);
			}

		}

	}

}
